package com.ipinyou.utils;

import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

public class HbaseUtils implements Serializable {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	public void hello() {
		System.out.println("nihao");
	}

	/**
	 * HAdmin init
	 * 
	 * @return
	 * @throws MasterNotRunningException
	 * @throws ZooKeeperConnectionException
	 * @throws IOException
	 */
	public HBaseAdmin getAdmin() throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
		Configuration config = HBaseConfiguration.create();
		HBaseAdmin admin = new HBaseAdmin(config);
		return admin;
	}

	/**
	 * 创建一个表
	 * 
	 * @param admin
	 * @throws IOException
	 */
	public void createTable(HBaseAdmin admin) throws IOException {
		if (admin.tableExists("t2")) {
			System.out.println("Table is exitst!");
		} else {
			HTableDescriptor descriptor = new HTableDescriptor("t2");
			descriptor.addFamily(new HColumnDescriptor("NB"));
			admin.createTable(descriptor);
			System.out.println("create success!!!");
		}

	}

	/**
	 * 查询所有的列表
	 * 
	 * @throws Exception
	 */
	public List getALLTables(HBaseAdmin admin) throws Exception {
		List<String> tables = null;
		if (admin != null) {
			HTableDescriptor[] listTables = admin.listTables();
			if (listTables.length > 0) {
				tables = new ArrayList<String>();
				for (HTableDescriptor tableDesc : listTables) {
					tables.add(tableDesc.getNameAsString());
					System.out.println(tableDesc.getNameAsString());
				}
			}
		}
		return tables;
	}

	/**
	 * 往表中添加一条记录
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param family
	 * @param column
	 * @param value
	 */
	public boolean addOneRecord(String tableName, String rowKey, String family, String column, byte[] value) {

		Configuration config = HBaseConfiguration.create();
		HTablePool hTable = new HTablePool(config, 1000);
		HTableInterface table = hTable.getTable(tableName);
		Put put = new Put(rowKey.getBytes());
		put.add(family.getBytes(), column.getBytes(), value);
		try {
			table.put(put);
			System.out.println("添加记录 " + rowKey + " 成功！");
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			System.out.println("添加记录 " + rowKey + " 成功！");
			return false;
		}
	}

	/**
	 * 查询一条记录
	 * 
	 * @param tableName
	 * @param rowKey
	 */
	public void getKey(String tableName, String rowKey) {

		Configuration config = HBaseConfiguration.create();
		HTablePool hTable = new HTablePool(config, 1000);
		HTableInterface table = hTable.getTable(tableName);
		Get get = new Get(rowKey.getBytes());
		try {
			Result rs = table.get(get);
			if (rs.raw().length == 0) {
				System.out.println("不存在关键字为 " + rowKey + " 的行！");
			} else {
				for (KeyValue kv : rs.raw()) {
					System.out.println(rowKey + "\t" + new String(kv.getValue()));
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 获取表的所有数据
	 * 
	 * @param tableName
	 * @throws Exception
	 */
	public void getAllData(String tableName) throws Exception {
		Configuration config = HBaseConfiguration.create();
		HTable hTable = new HTable(config, TableName.valueOf(tableName));
		Scan scan = new Scan();
		ResultScanner scanner = hTable.getScanner(scan);
		for (Result rs : scanner) {
			for (KeyValue kv : rs.raw()) {
				System.out.println(new String(kv.getKey()) + "\t" + new String(kv.getValue()));
			}
		}
	}

	/**
	 * 删除一条记录
	 * 
	 * @param tableName
	 * @param rowKey
	 */
	public void deleteRecord(String tableName, String rowKey) {
		Configuration config = HBaseConfiguration.create();
		HTablePool hTable = new HTablePool(config, 1000);
		HTableInterface table = hTable.getTable(tableName);
		Delete delete = new Delete(rowKey.getBytes());
		try {
			table.delete(delete);
			System.out.println("删除记录 " + rowKey + " 成功！");
		} catch (IOException e) {
			System.out.println("删除记录 " + rowKey + " 异常！");
			e.printStackTrace();
		}
	}

	/**
	 * 删除表
	 * 
	 * @param tableName
	 * @throws Exception
	 */
	public void deleteTable(String tableName, HBaseAdmin admin) {
		try {
			if (admin.tableExists(tableName)) {
				admin.disableTable(tableName);
				admin.deleteTable(tableName);
				System.out.println(tableName + " 删除成功！");
			}
		} catch (IOException e) {
			e.printStackTrace();
			System.out.println(tableName + " 删除失败！");
		}
	}

}
